/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


package org.apache.zookeeper.server.quorum;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.zookeeper.jmx.MBeanRegistry;
import org.apache.zookeeper.server.quorum.QuorumCnxManager.Message;
import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * Implementation of leader election using TCP. It uses an object of the class
 * QuorumCnxManager to manage connections. Otherwise, the algorithm is push-based
 * as with the other UDP implementations.
 *
 * There are a few parameters that can be tuned to change its behavior. First,
 * finalizeWait determines the amount of time to wait until deciding upon a leader.
 * This is part of the leader election algorithm.
 */


public class FastLeaderElection implements Election {
    private static final Logger LOG = LoggerFactory.getLogger(FastLeaderElection.class);

    /** 投票完成过半验证,之后需要等待接收队列后续消息的时长 */
    final static int finalizeWait = 200;
    /** 接收Notification的最大间隔时长  */
    final static int maxNotificationInterval = 60000;
    /** QuorumCnxManager对象，即选举leader时的IO管理器 */
    QuorumCnxManager manager;


    /**
     * Notifications are messages that let other peers know that
     * a given peer has changed its vote, either because it has
     * joined leader election or because it learned of another
     * peer with higher zxid or same zxid and higher server id
     */
    static public class Notification {
        /** Proposed leader */
        long leader;
        /** zxid of the proposed leader */
        long zxid;
        /** Epoch */
        long electionEpoch;
        /** current state of sender */
        QuorumPeer.ServerState state;
        /** Address of sender */
        long sid;
        /** epoch of the proposed leader */
        long peerEpoch;
    }

    /**
     * Messages that a peer wants to send to other peers.
     * These messages can be both Notifications and Acks
     * of reception of notification.
     */
    static public class ToSend {
        static enum mType {crequest, challenge, notification, ack}

        ToSend(mType type,
                long leader,
                long zxid,
                long electionEpoch,
                ServerState state,
                long sid,
                long peerEpoch) {

            this.leader = leader;
            this.zxid = zxid;
            this.electionEpoch = electionEpoch;
            this.state = state;
            this.sid = sid;
            this.peerEpoch = peerEpoch;
        }

        /** Proposed leader in the case of notification */
        long leader;

        /** id contains the tag for acks, and zxid for notifications */
        long zxid;

        /** Epoch */
        long electionEpoch;

        /** Current state; */
        QuorumPeer.ServerState state;

        /** Address of recipient */
        long sid;
        
        /** Leader epoch */
        long peerEpoch;
    }

    /**
     * ToSend队列，表示待发送消息的队列
     */
    LinkedBlockingQueue<ToSend> sendqueue;
    /**
     * Notification队列，表示接收消息的队列
     */
    LinkedBlockingQueue<Notification> recvqueue;

    /**
     * Multi-threaded implementation of message handler. Messenger
     * implements two sub-classes: WorkReceiver and  WorkSender. The
     * functionality of each is obvious from the name. Each of these
     * spawns a new thread.
     */

    private class Messenger {

        /**
         * Receives messages from instance of QuorumCnxManager on
         * method run(), and processes such messages.
         */
        class WorkerReceiver implements Runnable {
            volatile boolean stop;
            QuorumCnxManager manager;

            WorkerReceiver(QuorumCnxManager manager) {
                this.stop = false;
                this.manager = manager;
            }

            @Override
            public void run() {
                Message response;
                while (!stop) {
                    // Sleeps on receive
                    try{
                        response = manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
                        if(response == null) {
                            continue;
                        }

                        if(!self.getVotingView().containsKey(response.sid)){
                            Vote current = self.getCurrentVote();
                            ToSend notmsg = new ToSend(ToSend.mType.notification,
                                    current.getId(),
                                    current.getZxid(),
                                    logicalclock,
                                    self.getPeerState(),
                                    response.sid,
                                    current.getPeerEpoch());

                            sendqueue.offer(notmsg);
                        } else {
                            // Receive new message
                            // We check for 28 bytes for backward compatibility
                            if (response.buffer.capacity() < 28) {
                                LOG.error("Got a short response: " + response.buffer.capacity());
                                continue;
                            }
                            boolean backCompatibility = (response.buffer.capacity() == 28);
                            response.buffer.clear();

                            // State of peer that sent this message
                            QuorumPeer.ServerState ackstate = QuorumPeer.ServerState.LOOKING;
                            switch (response.buffer.getInt()) {
                            case 0:
                                ackstate = QuorumPeer.ServerState.LOOKING;
                                break;
                            case 1:
                                ackstate = QuorumPeer.ServerState.FOLLOWING;
                                break;
                            case 2:
                                ackstate = QuorumPeer.ServerState.LEADING;
                                break;
                            case 3:
                                ackstate = QuorumPeer.ServerState.OBSERVING;
                                break;
                            }

                            // Instantiate Notification and set its attributes
                            Notification n = new Notification();
                            n.leader = response.buffer.getLong();
                            n.zxid = response.buffer.getLong();
                            n.electionEpoch = response.buffer.getLong();
                            n.state = ackstate;
                            n.sid = response.sid;
                            if(!backCompatibility){
                                n.peerEpoch = response.buffer.getLong();
                            } else {
                                if(LOG.isInfoEnabled()){
                                    LOG.info("Backward compatibility mode, server id=" + n.sid);
                                }
                                n.peerEpoch = ZxidUtils.getEpochFromZxid(n.zxid);
                            }

                            /*
                             * Print notification info
                             */
                            if(LOG.isInfoEnabled()){
                                printNotification(n);
                            }

                            /*
                             * If this server is looking, then send proposed leader
                             */
                            if(self.getPeerState() == QuorumPeer.ServerState.LOOKING){
                                recvqueue.offer(n);

                                /*
                                 * Send a notification back if the peer that sent this
                                 * message is also looking and its logical clock is
                                 * lagging behind.
                                 */
                                if((ackstate == QuorumPeer.ServerState.LOOKING) && (n.electionEpoch < logicalclock)){
                                    Vote v = getVote();
                                    ToSend notmsg = new ToSend(ToSend.mType.notification,
                                            v.getId(),
                                            v.getZxid(),
                                            logicalclock,
                                            self.getPeerState(),
                                            response.sid,
                                            v.getPeerEpoch());
                                    sendqueue.offer(notmsg);
                                }
                            } else {
                                /*
                                 * If this server is not looking, but the one that sent the ack
                                 * is looking, then send back what it believes to be the leader.
                                 */
                                Vote current = self.getCurrentVote();
                                if(ackstate == QuorumPeer.ServerState.LOOKING){
                                    ToSend notmsg = new ToSend(
                                            ToSend.mType.notification,
                                            current.getId(),
                                            current.getZxid(),
                                            logicalclock,
                                            self.getPeerState(),
                                            response.sid,
                                            current.getPeerEpoch());
                                    sendqueue.offer(notmsg);
                                }
                            }
                        }
                    } catch (InterruptedException e) {
                        System.out.println("Interrupted Exception while waiting for new message" + e.toString());
                    }
                }
                LOG.info("WorkerReceiver is down");
            }
        }


        /**
         * 发送线程，不停轮询，把要发送的消息放到 queueSendMap里面
         */
        class WorkerSender implements Runnable {
            volatile boolean stop;
            QuorumCnxManager manager;

            WorkerSender(QuorumCnxManager manager){
                this.stop = false;
                this.manager = manager;
            }

            @Override
            public void run() {
                while (!stop) {
                    try {
                        ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS);
                        if(m == null) {
                            continue;
                        }

                        process(m);
                    } catch (InterruptedException e) {
                        break;
                    }
                }
                LOG.info("WorkerSender is down");
            }

            /**
             * Called by run() once there is a new message to send.
             *
             * @param m     message to send
             */
            private void process(ToSend m) {
                byte requestBytes[] = new byte[36];
                ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);

                /*
                 * Building notification packet to send
                 */
                requestBuffer.clear();
                requestBuffer.putInt(m.state.ordinal());
                requestBuffer.putLong(m.leader);
                requestBuffer.putLong(m.zxid);
                requestBuffer.putLong(m.electionEpoch);
                requestBuffer.putLong(m.peerEpoch);

                manager.toSend(m.sid, requestBuffer);

            }
        }

        /**
         * Test if both send and receive queues are empty.
         */
        public boolean queueEmpty() {
            return (sendqueue.isEmpty() || recvqueue.isEmpty());
        }


        WorkerSender ws;
        WorkerReceiver wr;

        /**
         * 初始化 WorkerSender  和 WorkerReceiver
         *
         * @param manager   Connection manager
         */
        Messenger(QuorumCnxManager manager) {
            this.ws = new WorkerSender(manager);
            Thread t = new Thread(this.ws, "WorkerSender[myid=" + self.getId() + "]");
            t.setDaemon(true);
            t.start();

            this.wr = new WorkerReceiver(manager);
            t = new Thread(this.wr, "WorkerReceiver[myid=" + self.getId() + "]");
            t.setDaemon(true);
            t.start();
        }

        /**
         * Stops instances of WorkerSender and WorkerReceiver
         */
        void halt(){
            this.ws.stop = true;
            this.wr.stop = true;
        }

    }

    /**
     * QuorumPeer对象,代表当前机器相关信息
     */
    QuorumPeer self;
    /**
     * 消息处理器，包含WorkerReceiver和WorkerSender两个内部类,处理发送队列和接受队列
     */
    Messenger messenger;
    /**
     * 逻辑时钟，相当于投票轮次
     */
    volatile long logicalclock;
    /**
     * 提议的leader
     */
    long proposedLeader;
    /**
     * 提议的leader的lastZxid
     */
    long proposedZxid;
    /**
     * 提议的leader的epoch
     */
    long proposedEpoch;


    /**
     * Returns the current vlue of the logical clock counter
     */
    public long getLogicalClock(){
        return logicalclock;
    }

    /**
     * Constructor of FastLeaderElection. It takes two parameters, one
     * is the QuorumPeer object that instantiated this object, and the other
     * is the connection manager. Such an object should be created only once
     * by each peer during an instance of the ZooKeeper service.
     *
     * @param self  QuorumPeer that created this object
     * @param manager   Connection manager
     */
    public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager){
        this.stop = false;
        this.manager = manager;
        starter(self, manager);
    }

    /**
     * This method is invoked by the constructor. Because it is a
     * part of the starting procedure of the object that must be on
     * any constructor of this class, it is probably best to keep as
     * a separate method. As we have a single constructor currently,
     * it is not strictly necessary to have it separate.
     *
     * @param self      QuorumPeer that created this object
     * @param manager   Connection manager
     */
    private void starter(QuorumPeer self, QuorumCnxManager manager) {
        this.self = self;
        proposedLeader = -1;
        proposedZxid = -1;

        sendqueue = new LinkedBlockingQueue<ToSend>();
        recvqueue = new LinkedBlockingQueue<Notification>();
        this.messenger = new Messenger(manager);  // 初始化 Messager
    }

    private void leaveInstance(Vote v) {
        recvqueue.clear();
    }

    public QuorumCnxManager getCnxManager(){
        return manager;
    }

    volatile boolean stop;

    @Override
    public void shutdown(){
        stop = true;
        LOG.debug("Shutting down connection manager");
        manager.halt();
        LOG.debug("Shutting down messenger");
        messenger.halt();
        LOG.debug("FLE is down");
    }


    /**
     * 会对所有参与选举的server端发送自己的选票
     */
    private void sendNotifications() {
        for (QuorumServer server : self.getVotingView().values()) {
            long sid = server.id;

            // 对其他所有机器发送消息
            ToSend notmsg = new ToSend(
                    ToSend.mType.notification,  // 投票类型为notification
                    proposedLeader,             // 投票leader的myid值
                    proposedZxid,               // 投票leader的zxid值
                    logicalclock,               // 当前选举轮次
                    QuorumPeer.ServerState.LOOKING, // 当前服务器状态
                    sid,                        // myid
                    proposedEpoch);              // 当前currentEpoch的值，即currentEpoch文件的值

            sendqueue.offer(notmsg);
        }
    }


    private void printNotification(Notification n){
        LOG.info("Notification: " + n.leader + " (n.leader), 0x"
                + Long.toHexString(n.zxid) + " (n.zxid), 0x"
                + Long.toHexString(n.electionEpoch) + " (n.round), " + n.state
                + " (n.state), " + n.sid + " (n.sid), 0x"
                + Long.toHexString(n.peerEpoch) + " (n.peerEPoch), "
                + self.getPeerState() + " (my state)");
    }

    /**
     * 比较两个投票中，哪张投票能"赢"
     * 选票PK， 满足一个即可
     * 1- 选举轮次更大
     * 2- zxid，事务ID更大
     * 3- server id ，即myid 更大
     */
    protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
        if(self.getQuorumVerifier().getWeight(newId) == 0){
            return false;
        }

        return ((newEpoch > curEpoch) ||
                ((newEpoch == curEpoch) &&
                ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId)))));
    }

    /**
     * 验证自己的投票是否通过了集群验证器的验证(通常是过半)
     * @param votes 收到的投票提议集合
     * @param vote  自己的投票
     * @return
     */
    private boolean termPredicate( HashMap<Long, Vote> votes, Vote vote) {

        HashSet<Long> set = new HashSet<>();
        // 投票集合中，投的票和自己票一样的筛选出来，写入 set里, 看是否过半
        for (Map.Entry<Long,Vote> entry : votes.entrySet()) {
            if (vote.equals(entry.getValue())){
                set.add(entry.getKey());
            }
        }
        // 调用的是 QuorumMaj 实现类
        return self.getQuorumVerifier().containsQuorum(set);
    }

    /**
     * 用于接收某些状态已经为leading和following的消息时(即部分server认为leader已经选举出来了，自己还在looking状态下),检验leader的有效性
     *
     * @param votes set of votes
     * @param   leader  leader id
     * @param   electionEpoch   epoch id
     */
    private boolean checkLeader( HashMap<Long, Vote> votes, long leader, long electionEpoch){

        boolean predicate = true;

        // 自己不为leader
        if(leader != self.getId()){
            if(votes.get(leader) == null) {
                // 投票记录中，没有来自leader的投票
                predicate = false;
            }
            // leader不知道自己是leader
            else if(votes.get(leader).getState() != ServerState.LEADING) {
                predicate = false;
            }
        }
        // 如果大家认为我是leader，但是逻辑时钟不等于选举周期
        else if(logicalclock != electionEpoch) {
            predicate = false;
        } 
        return predicate;
    }

    synchronized void updateProposal(long leader, long zxid, long epoch){
        proposedLeader = leader;
        proposedZxid = zxid;
        proposedEpoch = epoch;
    }

    synchronized Vote getVote(){
        return new Vote(proposedLeader, proposedZxid, proposedEpoch);
    }

    /**
     * A learning state can be either FOLLOWING or OBSERVING.
     * This method simply decides which one depending on the
     * role of the server.
     *
     * @return ServerState
     */
    private ServerState learningState(){
        if(self.getLearnerType() == LearnerType.PARTICIPANT){
            LOG.debug("I'm a participant: " + self.getId());
            return ServerState.FOLLOWING;
        }
        else{
            LOG.debug("I'm an observer: " + self.getId());
            return ServerState.OBSERVING;
        }
    }

    /**
     * Returns the initial vote value of server identifier.
     *
     * @return long
     */
    private long getInitId(){
        if(self.getLearnerType() == LearnerType.PARTICIPANT)
            return self.getId();
        else return Long.MIN_VALUE;
    }

    /**
     * Returns initial last logged zxid.
     *
     * @return long
     */
    private long getInitLastLoggedZxid(){
        if(self.getLearnerType() == LearnerType.PARTICIPANT)
            return self.getLastLoggedZxid();
        else return Long.MIN_VALUE;
    }

    /**
     * Returns the initial vote value of the peer epoch.
     *
     * @return long
     */
    private long getPeerEpoch(){
        if(self.getLearnerType() == LearnerType.PARTICIPANT)
        	try {
        		return self.getCurrentEpoch();
        	} catch(IOException e) {
        		RuntimeException re = new RuntimeException(e.getMessage());
        		re.setStackTrace(e.getStackTrace());
        		throw re;
        	}
        else return Long.MIN_VALUE;
    }
    
    /**
     * Starts a new round of leader election. Whenever our QuorumPeer
     * changes its state to LOOKING, this method is invoked, and it
     * sends notifications to all other peers.
     */
    @Override
    public Vote lookForLeader() throws InterruptedException {
        // jmx代码，忽略
        try {
            self.jmxLeaderElectionBean = new LeaderElectionBean();
            MBeanRegistry.getInstance().register( self.jmxLeaderElectionBean, self.jmxLocalPeerBean);
        } catch (Exception e) {
            LOG.warn("Failed to register with JMX", e);
            self.jmxLeaderElectionBean = null;
        }

        if (self.start_fle == 0) {
            // 开始fast leader election的时间
           self.start_fle = System.currentTimeMillis();
        }

        try {
            //本轮次logicalclock, 接收到的投票集合。
            // 用于记录当前服务器在本轮次的选举中收到的所有外部投票
            HashMap<Long, Vote> recvset = new HashMap< >();
            // 选举之外的投票集合(即对方为following和leading状态)
            HashMap<Long, Vote> outofelection = new HashMap< >();
            int notTimeout = finalizeWait;

            synchronized(this){
                logicalclock++; // 轮次+1
                // 更新提议，包含(myid,lastZxid,epoch)
                // 第一次选举前，每台服务器都会将自己推举为leader
                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
            }

            sendNotifications(); // 发起 自己的投票 给别人， 通过QuorumCnxManager
            // 通过 sendqueue 发送出去

            // 只要没有停止，并且状态处于LOOKING状态
            while ((self.getPeerState() == ServerState.LOOKING) && (!stop)){
                // 通过 recvqueue 接收其他机器的选票
                Notification n = recvqueue.poll(notTimeout, TimeUnit.MILLISECONDS); //在200ms内接收通知

                // 如果没有接收到外部投票，
                if(n == null){
                    // QuorumCnxManager.queueSendMap为空，则重新发送自己的投票
                    if(manager.haveDelivered()){
                        //重新发送通知
                        sendNotifications();
                    }
                    else {
                        // 检查连接，没有连接的话重新和有vote资格的sid的服务器创建连接
                        // 同步连接上所有有vote资格的sid
                        manager.connectAll();
                    }
                    int tmpTimeOut = notTimeout*2;
                    notTimeout = (tmpTimeOut < maxNotificationInterval? tmpTimeOut : maxNotificationInterval);
                    LOG.info("Notification time out: " + notTimeout);
                }
                else if(self.getVotingView().containsKey(n.sid)) {
                    // 查看 其他机器的状态
                    switch (n.state) {
                    case LOOKING:
                        //如果接收的epoch比自己的高，如果外部投票的轮次大于内部投票的轮次
                        if (n.electionEpoch > logicalclock) {
                            // 立即更新自己的选举轮次
                            logicalclock = n.electionEpoch;
                            //之前接收到的投票作废，必须同一个epoch才行。
                            // 清空所有已经收到的投票
                            recvset.clear();

                            // 如果接收的Notification可以替代自己的提议
                            if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch, getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
                                // 那么就更新自己的提议
                                updateProposal(n.leader, n.zxid, n.peerEpoch);
                            } else {
                                //否则还原为初始化提议
                                updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
                            }
                            //给各server发送通知
                            sendNotifications();
                        }
                        //如果接受的epoch比自己的低,就不用管
                        else if (n.electionEpoch < logicalclock) {
                            break;
                        }
                        // 外部投票的选举轮次和内部投票一致，开始pk选票
                        // totalOrderPredicate 选票PK核心算法，依次比较 选举轮次，zxid, sid（myid）
                        else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,  proposedLeader, proposedZxid, proposedEpoch)) {
                            // 如果其他机器胜出，更新自己提议，并将变更发送出去
                            updateProposal(n.leader, n.zxid, n.peerEpoch);
                            sendNotifications();
                        }
                        // recvset用于记录当前服务器在本轮次的leader选举中收到的所有外部投票，按照SID来区分
                        recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));

                        //如果集群验证器验证通过 当前机器是"leader"(通常是过半机器投票给当前机器) ，即自己的投票超过了大多数
                        if (termPredicate(recvset, new Vote(proposedLeader, proposedZxid, logicalclock, proposedEpoch))) {
                            // 如果还能接收到其他服务器的投票
                            while((n = recvqueue.poll(finalizeWait, TimeUnit.MILLISECONDS)) != null){
                                if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,  proposedLeader, proposedZxid, proposedEpoch)){
                                    // 如果其他服务器的投票 PK 赢了 自己的投票，跳出上一行的while, 继续走大的while循环
                                    recvqueue.put(n);
                                    break;
                                }
                            }
                            // 如果不再接收到其他选票，
                            // 说明当前服务器的leader选票就是最终leader的SID(不一定是自己),也就是有过半的服务器认可了当前的内部投票
                            if (n == null) {
                                // 因为已经有leader了，自己肯定是投给leader的,
                                // 如果 自己的投票是投给了自己，自己就是leader, 否则，你就是follower 或者 observer
                                self.setPeerState((proposedLeader == self.getId()) ? ServerState.LEADING: learningState());
                                Vote endVote = new Vote(proposedLeader, proposedZxid, proposedEpoch); // debug用
                                // 清空接收消息队列
                                leaveInstance(endVote);
                                // 返回投票信息
                                return endVote;
                            }
                        }
                        break;
                    case OBSERVING:
                        LOG.debug("Notification from observer: " + n.sid);
                        break;
                    case FOLLOWING: // 如果其他机器已经是leader状态了
                    case LEADING:
                        // 如果投票轮次 和 自己一样
                        if(n.electionEpoch == logicalclock){
                            //同一个选举周期，添加记录
                            recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
                            //是否能够加入已有的集群
                            if(termPredicate(recvset, new Vote(n.leader,
                                            n.zxid, n.electionEpoch, n.peerEpoch, n.state))
                                            && checkLeader(outofelection, n.leader, n.electionEpoch)) {
                                // 其他节点的id 是否是自己的id,如果不是，自己follower状态
                                self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING: learningState());

                                Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch);
                                // 清空 接收消息队列
                                leaveInstance(endVote);
                                return endVote;
                            }
                        }

                        // 选举轮次不一样，自己的投票轮次可能落后了，因为leader的投票轮次一定是最大的，否则不可能选举成leder
                        outofelection.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state));

                        if (termPredicate(outofelection, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch, n.state))
                                && checkLeader(outofelection, n.leader, n.electionEpoch)) {
                            // 修改自己的选举轮次
                            synchronized(this){
                                logicalclock = n.electionEpoch;
                                self.setPeerState((n.leader == self.getId()) ? ServerState.LEADING: learningState());
                            }
                            Vote endVote = new Vote(n.leader, n.zxid, n.peerEpoch);
                            leaveInstance(endVote);
                            return endVote;
                        }
                        break;
                    default:
                        LOG.warn("Notification state unrecoginized: " + n.state + " (n.state), " + n.sid + " (n.sid)");
                        break;
                    }
                } else {
                    LOG.warn("Ignoring notification from non-cluster member " + n.sid);
                }
            }
            return null;
        } finally {
            try {
                if(self.jmxLeaderElectionBean != null){
                    MBeanRegistry.getInstance().unregister( self.jmxLeaderElectionBean);
                }
            } catch (Exception e) {
                LOG.warn("Failed to unregister with JMX", e);
            }
            self.jmxLeaderElectionBean = null;
        }
    }
}
